package com.yeeiee;

import com.yeeiee.config.JdbcCatalogConfig;
import com.yeeiee.utils.FactoryUtils;
import lombok.val;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;

public class CatalogTest1 {
    public static void main(String[] args) throws TableNotExistException, TableAlreadyExistException, DatabaseNotExistException {
        val env = StreamExecutionEnvironment.getExecutionEnvironment();
        val tabEnv = StreamTableEnvironment.create(env);

        // mysql catalog
        val configuration = new Configuration();
        configuration.set(JdbcCatalogConfig.url, "jdbc:mysql://localhost:3306/chen_test");
        configuration.set(JdbcCatalogConfig.username, "root");
        configuration.set(JdbcCatalogConfig.password, "123");
        val catalog = FactoryUtils.getCatalog("mysql", configuration);
        tabEnv.registerCatalog("mysql", catalog);


        tabEnv.executeSql("create table if not exists mysql.chen_test.t_user_1 like mysql.chen_test.t_user");

        tabEnv.executeSql("insert into mysql.chen_test.t_user_1 select * from mysql.chen_test.t_user");

//        tabEnv.executeSql("drop database if exists mysql.chen");
    }
}
